package com.qdu.average;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;

import java.io.BufferedReader;
import java.io.FileWriter;
import java.io.InputStreamReader;
import java.net.URI;

public class PC_HDFSReader {

    public static void readAndWriteJson(String hdfsUri, String outputPath, String localJsonPath) {
        Configuration conf = new Configuration();
        FileSystem fs = null;
        BufferedReader br = null;
        FileWriter fileWriter = null;

        try {
            fs = FileSystem.get(new URI(hdfsUri), conf);
            Path path = new Path(outputPath);

            br = new BufferedReader(new InputStreamReader(fs.open(path)));

            ObjectMapper mapper = new ObjectMapper();
            ArrayNode jsonArray = mapper.createArrayNode();

            String line;
            while ((line = br.readLine()) != null) {
                String[] fields = line.split("\t");
                ObjectNode jsonObject = mapper.createObjectNode();
                jsonObject.put("user_id", fields[0]);
                jsonObject.put("average_time", Integer.parseInt(fields[1]));

                jsonArray.add(jsonObject);
            }

            fileWriter = new FileWriter(localJsonPath);
            mapper.writerWithDefaultPrettyPrinter().writeValue(fileWriter, jsonArray);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            IOUtils.closeStream(br);
            IOUtils.closeStream(fileWriter);
        }
    }

    public static void main(String[] args) {
        String hdfsUri = "hdfs://songchao:9000";
        String outputPath = "/output/average/part-r-00000";

//        String localJsonPath = "purchase_conversion_rate.json";
        String localJsonPath = "src/main/resources/average.json";


        readAndWriteJson(hdfsUri, outputPath, localJsonPath);
    }
}
